package com.jxtest.studykafka.kafkatest.rest;

import com.alibaba.fastjson.JSON;
import com.jxtest.studykafka.kafkatest.constant.KafkaConst;
import com.jxtest.studykafka.kafkatest.entity.UserLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @className: KafkaRest
 * @description: 测试kafka
 * @author: jiaxu
 * @create: 2021/6/16
 *
 * 安装：https://blog.csdn.net/weixin_39626409/article/details/110289099?spm=1001.2101.3001.6650.2&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EOPENSEARCH%7ERate-2-110289099-blog-133034518.235%5Ev38%5Epc_relevant_sort_base3&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EOPENSEARCH%7ERate-2-110289099-blog-133034518.235%5Ev38%5Epc_relevant_sort_base3&utm_relevant_index=1
 * 命令：https://blog.csdn.net/YJ000312/article/details/129832279
 *
 * ./kafka-topics.sh --zookeeper localhost:2181 --list
 * ./kafka-console-producer.sh --broker-list localhost:9092 --topic test
 *
 * ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
 *
 *
 * 1、启动kafka服务
 *
 * -- 启动zookeeper
 * nohup bin/zookeeper-server-start.sh config/zookeeper.properties &
 * -- 启动kafka
 * bin/kafka-server-start.sh config/server.properties &
 *
 * 2、停止kafka服务
 * ./kafka-server-stop.sh
 *
 * 3、查看所有的话题
 * ./kafka-topics.sh --list --zookeeper localhost:9092
 *
 * 4、查看所有话题的详细信息
 * ./kafka-topics.sh --zookeeper localhost:2181 --describe
 *
 * 5、列出指定话题的详细信息
 * ./kafka-topics.sh --zookeeper localhost:2181 --describe  --topic demo
 *
 * 6、删除一个话题
 * ./kafka-topics.sh --zookeeper localhost:2181 --delete  --topic test
 *
 * 7、创建一个叫test的话题，有两个分区，每个分区3个副本
 * ./kafka-topics.sh --zookeeper localhost:2181 --create --topic test --replication-factor 3 --partitions 2
 *
 *  8、测试kafka发送和接收消息（启动两个终端）
 * #发送消息（注意端口号为配置文件里面的端口号）
 * ./kafka-console-producer.sh --broker-list localhost:9092 --topic test
 * #消费消息（可能端口号与配置文件保持一致，或与发送端口保持一致）
 * ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning   #加了--from-beginning 重头消费所有的消息
 * ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test         #不加--from-beginning 从最新的一条消息开始消费
 *
 * 9、查看某个topic对应的消息数量
 * ./kafka-run-class.sh  kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -1
 *
 * 10、显示所有消费者
 * ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
 *
 * 11、获取正在消费的topic（console-consumer-63307）的group的offset
 * ./kafka-consumer-groups.sh --describe --group console-consumer-63307 --bootstrap-server localhost:9092
 *
 * 11、显示消费者
 * ./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
 **/
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaRest {
    @RequestMapping("/test/{name}")
    public String testHello(@PathVariable("name") String name){
        return "hello " + name;
    }

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送数据
     * @param userId
     */
    @RequestMapping("/sendLog/{userId}")
    public String sendLog(@PathVariable("userId") String userId){
        UserLog userLog = new UserLog();
        userLog.setUsername("jhp").setUserid(userId).setState("0");
        log.info("发送用户日志数据:"+userLog);
        kafkaTemplate.send(KafkaConst.PRODUCER_TOPIC, JSON.toJSONString(userLog));
        return "发送成功";
    }


    /**
     * 发送过车测试数据
     * @param
     */
    @RequestMapping("/sendPassCar")
    public String sendPassCar(){
        String passCarData = "1234VERSION=1.0\n" +
                "PASSTIME=2019-12-13 11:32:51 988\n" +
                "CARSTATE=1\n" +
                "CARPLATE=鲁A9K6J8\n" +
                "PLATETYPE=2\n" +
                "SPEED=116\n" +
                "PLATECOLOR=2\n" +
                "LOCATIONID=-1\n" +
                "DEVICEID=-1\n" +
                "DRIVEWAY=2\n" +
                "DRIVEDIR=02\n" +
                "CAPTUREDIR=2\n" +
                "CARCOLOR=2\n" +
                "CARBRAND=99\n" +
                "CARBRANDZW=其他\n" +
                "TGSID=1031\n" +
                "PLATECOORD=783,1362,98,24\n" +
                "CABCOORD=0,0,0,0\n" +
                "IMGID1=http://20.0.51.15:9099/image/fastdfs/group50/M00/8D/25/FAA2al3zBmWAUuAwAAdcKEpzI2k432.jpg\n" +
                "IMGID2=\n" +
                "IMGID3=http://20.0.51.15:9099/image/fastdfs/group59/M00/8C/E1/FAA2PF3zBmWAbnoqAAAK5LoywR0550.jpg\n" +
                "IMGID4=\n" +
                "IMGID5=";
        //log.info("发送过车数据:" + passCarData);
        kafkaTemplate.send(new ProducerRecord(KafkaConst.PRODUCER_TOPIC,JSON.toJSONString(passCarData).getBytes()));
        return "发送成功";
    }

}
